package f8watermark;

import orderdata.OrderData;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;

import java.sql.Date;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;

public class VMSource implements SourceFunction<OrderData> {
    private Boolean running = true;

    @Override
    public void run(SourceContext<OrderData> sourceContext) throws Exception {
        long orderId = 0;
        Random random = new Random();
        String[] categories = {"Electronics", "Clothing", "Books", "Furniture", "Food"};
        String[] regions = {"North America", "Europe", "Asia", "Africa", "Australia"};
        // 用于记录最大时间戳（模拟水位线机制）
        long maxTimestamp = Long.MIN_VALUE;
        while (running) {
            long randomMillis = ThreadLocalRandom.current().nextLong(
                    new Date(2000 - 1900, 0, 1).getTime(),
                    new Date(2025 - 1900, 11, 31).getTime()
            );
            // 创建一个订单对象并发送到下游
            OrderData order = new OrderData(
                    randomMillis, // 订单日期
                    categories[random.nextInt(categories.length)], // 商品品类
                    regions[random.nextInt(regions.length)], // 销售地区
                    100 + random.nextInt(1000), // 用户 ID
                    1 + random.nextInt(5), // 订单数
                    50 + random.nextDouble() * 450 // 订单金额（50 到 500 之间）

            );
            // 使用 collectWithTimestamp 方法将数据发送出去，并指明数据中的时间戳字段
            sourceContext.collectWithTimestamp(order, order.order_date);  // order_date 字段为时间戳
            // 更新最大时间戳
            maxTimestamp = Math.max(maxTimestamp, randomMillis);
            Thread.sleep(1000); // 每秒生成一条数据
            // 发送水位线 (模拟水位线发出机制)
            // 水位线是最大时间戳减去一个小的偏移量（此处为 1 毫秒）
            sourceContext.emitWatermark(new Watermark(maxTimestamp - 1L));
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}
